/**
 * 
 */

package com.wutianyi.taskqueue.queue.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import com.wutianyi.taskqueue.dao.TaskDao;
import com.wutianyi.taskqueue.queue.Task;
import com.wutianyi.taskqueue.queue.TaskQueue;
import com.wutianyi.taskqueue.queue.persistent.DBPersistent;
import com.wutianyi.taskqueue.utils.ThreadUtils;

/**
 * @author wutianyi
 * 
 */
public class TaskQueueImpl<T extends Task> implements TaskQueue<T> {

	private static Logger					logger					=
																			Logger
																					.getLogger(TaskQueueImpl.class);
	/**
	 * 最大的segment数
	 */
	private static int						MAX_SEGMENT_SIZE		= 5;
	private static int						MAX_EACH_SEGMENT_SIZE	= 100000;
	/**
	 * 用于删除任务的线程池的最大数
	 */
	private static int						MAX_DELETE_TASK_THREAD	= 5;

	private int								segmentSize				= 0;
	private int								eachSegmentSize			= 0;

	private Segment<T>[]					segments;
	/**
	 * 保存超过segment的task
	 * 由用户自己设置，如果没有设置，那么超过的task将丢弃
	 * 对于处理失败的任务，会放入这个queue中，所以这个对象不能为空
	 */
	private TaskQueue<T>					outOfSizeQueue;

	/**
	 * outOfSizeQueue的索引
	 */
	private int								outOfSizeQueueIndex		= 0;

	private ReentrantLock					locks					= new ReentrantLock();

	private LinkedBlockingQueue<Integer>	outputIndex				=
																			new LinkedBlockingQueue<Integer>();
	/**
	 *表示正在处理的tasks，会定时扫面任务是否处理超时
	 */
	// private CopyOnWriteArrayList<T> dealingTasks = new CopyOnWriteArrayList<T>();

	private ConcurrentHashMap<String, T>	dealingTasks			=
																			new ConcurrentHashMap<String, T>();

	private TaskDao<T>						taskDao;
	/**
	 * 定时回收超时处理的任务
	 */
	private Thread							recoverThread;

	private DBPersistent<T>					persistent;

	/**
	 * 删除任务的线程数
	 */
	private int								deleteTaskThread;
	/**
	 * 保存删除的任务的对象
	 */
	private BlockingQueue<DeleteTaskDO>		deleteTaskQueue;

	private Thread[]						deleteThreads;

	// FIXME 删除
	public AtomicInteger					deleteCount				= new AtomicInteger(0);
	public AtomicInteger					deleteFromIndex			= new AtomicInteger(0);

	public TaskQueueImpl(int segmentSize, int eachSegmentSize, int deleteTaskThread) {

		this.segmentSize = segmentSize;
		this.eachSegmentSize = eachSegmentSize;
		this.outOfSizeQueueIndex = segmentSize;
		this.deleteTaskThread = deleteTaskThread;

		this.deleteTaskQueue = new LinkedBlockingQueue<DeleteTaskDO>();

		segments = new Segment[segmentSize];
		for(int i = 0; i < segmentSize; i++) {
			segments[i] = new Segment<T>();
		}
		recoverThread = new Thread(new Runnable() {

			@Override
			public void run() {

				while(true) {

					for(Entry<String, T> task : dealingTasks.entrySet()) {
						if(task.getValue().isOutOfTime(System.currentTimeMillis())) {
							if(dealingTasks.containsKey(task.getKey())) {
								wakeUpTask(task.getValue());
								dealingTasks.remove(task.getKey());
							}
						}
					}

					try {
						Thread.sleep(5000);
					}catch(InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		});
		recoverThread.start();
		initDeleteThread();
	}

	public TaskQueueImpl() {

		this(MAX_SEGMENT_SIZE, MAX_EACH_SEGMENT_SIZE, MAX_DELETE_TASK_THREAD);
	}

	public void shutdown() {

		if(null != recoverThread && recoverThread.isAlive()) {
			recoverThread.interrupt();
		}
		for(Thread thread : deleteThreads) {
			if(null != thread && thread.isAlive()) {
				thread.interrupt();
			}
		}
		try {
			for(Thread thread : deleteThreads) {
				if(null != thread && thread.isAlive()) {
					recoverThread.join();
				}
			}
			recoverThread.join();
		}catch(InterruptedException e) {
			e.printStackTrace();
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#drainTo(java.util.Collection, int)
	 */
	@Override
	public int drainTo(Collection<T> c, int size) {

		return 0;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#offer(java.lang.Object)
	 */
	@Override
	public boolean offer(T t) {

		int index = Math.abs(t.getTaskId().hashCode()) % segmentSize;

		if(segments[index].size() >= eachSegmentSize) {
			index = this.outOfSizeQueueIndex;

			if(null != outOfSizeQueue) {
				outputIndex.offer(index);
				return outOfSizeQueue.offer(t);
			}
		}
		outputIndex.offer(index);
		return segments[index].offer(t);
	}

	/**
	 * 唤醒哪些处理超时的任务
	 * 处理超时的，放入outOfQueue中
	 * 哪些超时的任务的优先级
	 * 
	 * @param task
	 */
	public void wakeUpTask(T task) {

		if(null != outOfSizeQueue) {
			// logger.info("wakeUpTask: " + task.getTaskId());
			outputIndex.offer(this.outOfSizeQueueIndex);
			outOfSizeQueue.offer(task);
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#poll()
	 */
	@Override
	public T poll(int hideTime) {

		if(outputIndex.size() == 0) {
			return null;
		}

		Integer index = outputIndex.poll();
		if(null == index) {
			return null;
		}
		T t = null;
		if(index == outOfSizeQueueIndex) {
			if(null != outOfSizeQueue) {
				t = outOfSizeQueue.poll(hideTime);
				if(null == t) {
					// FIXME 当在outOfSizeQueue中的时候，但是却获取不到任务，说明持久化还没有完成，将任务index放回outputIndex中
					outputIndex.offer(index);
				}
			}
		}else {

			t = segments[index].poll(hideTime);
		}
		if(null == t) {
			return null;
		}
		t.setBeginDealTime(System.currentTimeMillis());
		t.setHideTime(hideTime);
		dealingTasks.put(t.getTaskId(), t);
		// dealingTasks.add(t);
		return t;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#size()
	 */
	@Override
	public int size() {

		int size = 0;
		try {
			locks.lock();

			for(Segment<T> segment : segments) {
				logger.info("segment_size: " + segment.size());
				size += segment.size();
			}
			if(null != outOfSizeQueue) {
				size += outOfSizeQueue.size();
				logger.info("out_of_size_queue size: " + outOfSizeQueue.size());
			}
			logger.info("输出队列数量：" + this.outputIndexSize());
			logger.info("正在处理的任务的数量：" + this.dealingTasks.size());
			logger.info("删除数据库任务的个数：" + deleteCount.get());
			logger.info("删除内存索引的任务的个数：" + deleteFromIndex.get());
		}finally {
			locks.unlock();
		}
		return size;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#drainTo(int)
	 */
	@Override
	public List<T> drainTo(int size, int hideTime) {

		if(0 == size || 0 == outputIndex.size()) {
			return null;
		}
		List<T> tasks = new ArrayList<T>();

		for(int i = 0; i < size; i++) {
			T t = poll(hideTime);
			if(null == t) {
				break;
			}
			tasks.add(t);
		}

		return tasks;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.wutianyi.queue.TaskQueue#batchAddTask(java.util.List)
	 */
	@Override
	public void batchAddTask(List<T> tasks) {

		if(null == tasks || 0 == tasks.size()) {
			return;
		}
		for(T t : tasks) {
			offer(t);
		}
	}

	public int outputIndexSize() {

		return outputIndex.size();
	}

	public TaskQueue<T> getOutOfSizeQueue() {

		return outOfSizeQueue;
	}

	public void setOutOfSizeQueue(TaskQueue<T> outOfSizeQueue) {

		this.outOfSizeQueue = outOfSizeQueue;
	}

	public int getEachSegmentSize() {

		return eachSegmentSize;
	}

	public void setPersistent(DBPersistent<T> persistent) {

		this.persistent = persistent;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.duowan.zone.taskqueue.queue.TaskQueue#removeTask(java.lang.String)
	 */
	@Override
	public boolean removeTask(T t) {

		T task = dealingTasks.remove(t.getTaskId());
		boolean result = false;
		if(null != task) {
			result = true;
			boolean rResult = persistent.removeTaskFromIndex(t);
			if(!rResult) {
				deleteCount.getAndIncrement();
				// logger.info("删除的任务的ID：" + t.getTaskId());
				deleteTaskQueue.offer(new DeleteTaskDO(t.getTaskId(), 0));
			}else {
				deleteFromIndex.getAndIncrement();
			}
		}
		return result;
	}

	public void setTaskDao(TaskDao<T> taskDao) {

		this.taskDao = taskDao;
	}

	private void initDeleteThread() {

		deleteThreads = new Thread[deleteTaskThread];
		DeleteListener listener = new DeleteListener();
		for(Thread thread : deleteThreads) {
			thread = new Thread(listener);
			thread.start();
		}
	}

	private class DeleteListener implements Runnable {

		/*
		 * (non-Javadoc)
		 * 
		 * @see java.lang.Runnable#run()
		 */
		@Override
		public void run() {

			while(true) {
				try {
					DeleteTaskDO task = deleteTaskQueue.take();
					if(null != task) {
						int result = taskDao.removeTask(task.getTaskId());
						if(result == 0 && task.getDeleteCount() <= 3) {
							task.setDeleteCount(task.getDeleteCount() + 1);
							deleteTaskQueue.offer(task);
							logger.info("Delete fail!");
							if(deleteTaskQueue.size() < 10) {
								ThreadUtils.sleep(100);
							}
						}
						if(result == 1) {
							logger.info("Delete Success: " + task.getTaskId());
						}
					}
				}catch(InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}

	}

	public static void main(String[] args) throws InterruptedException {

	}

}
